package com.ipinyou.mprofile

import java.lang.String

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD.rddToPairRDDFunctions
import org.apache.spark.streaming.Minutes
import org.apache.spark.streaming.Seconds
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.dstream.DStream.toPairDStreamFunctions
import org.apache.spark.streaming.kafka.KafkaUtils
import org.slf4j.LoggerFactory
import org.slf4j.MarkerFactory

import com.ipinyou.hbase.model.MUser
import com.ipinyou.hbase.service.impl.HBaseConnectionServiceImpl

import kafka.serializer.StringDecoder
import scala.collection.mutable.ArrayBuffer
import org.apache.spark.storage.StorageLevel

/**
 * Created by miaoyujia on 15/12/1.
 */

object MbScoringTest2 {

  var sc: SparkContext = null

  // log
  val log = LoggerFactory.getLogger(this.getClass)
  val hbaseServerce = new HBaseConnectionServiceImpl();
  val PARTITIONS = 1050

  /**
   * Spark批量读Hbase
   */
  def getFromHbaseByBatch(iter: Iterator[(String, Float)]): Iterator[(String, Float)] = {
    val muser = new MUser()
    val emMap = new scala.collection.mutable.HashMap[String, Float]

    while (iter.hasNext) {
      val arr = iter.next()
      val id = arr._1
      val sweight = arr._2
      emMap(id) = sweight
    }
    val idlist = emMap.keySet.toArray
    val rlist = hbaseServerce.get(muser, idlist: _*)

    for (i <- 0 until rlist.size()) {
      val ruser = rlist.get(i)
      val uid = ruser.getId
      val weight = ruser.getContent.toFloat

      val sweight = emMap.getOrElse(uid, 0.toFloat)
      emMap(uid) = weight + sweight
    }
    emMap.toArray.iterator
  }

  def existsAll(iter: Iterator[(String, Float)]): Iterator[(String, Float)] = {
    val muser = new MUser()
    val emMap = new scala.collection.mutable.HashMap[String, Float]

    while (iter.hasNext) {
      val arr = iter.next()
      val id = arr._1
      val sweight = arr._2
      emMap(id) = sweight
    }
    val idlist = emMap.keySet.toArray
    val rlist = hbaseServerce.isExist(muser, idlist: _*)

    var i = 0;
    for ((k, v) <- emMap.toIterator) {
      var flag: Float = 0
      if (rlist(i) == true) flag = 1
      emMap(k) = flag
      i += 1
    }
    emMap.toArray.iterator
  }

  def exists(iter: Iterator[(String, Float)]): Iterator[(String, Float)] = {
    val muser = new MUser()
    val emMap = new scala.collection.mutable.HashMap[String, Float]

    while (iter.hasNext) {
      val arr = iter.next()
      val id = arr._1
      val sweight = arr._2
      emMap(id) = sweight
    }
    val idlist = emMap.keySet.toArray
    val rlist = hbaseServerce.isExist(muser, idlist: _*)

    var i = 0;
    for ((k, v) <- emMap.toIterator) {
      var flag: Float = 0
      if (rlist(i) == true) flag = 1
      emMap(k) = flag
      i += 1
    }
    emMap.toArray.iterator
  }

  def save2HbaseByBatch(iter: Iterator[(String, Float)]) {
    val musers = new ArrayBuffer[MUser]()
    while (iter.hasNext) {
      val muser = new MUser()
      val arr = iter.next()
      val rowkey = arr._1
      val weight = arr._2
      muser.setId(rowkey)
      muser.setContent(weight.toString)
      musers.append(muser)
    }
    hbaseServerce.insert(musers: _*)
  }

  def main(args: Array[String]): Unit = {

    // local field

    // paremeters
    var model_file = ""
    var basefeat2extentfeat_file = ""
    var intervel = 0
    val config = new SparkConf().setAppName("mobile.scoring")
    // spark  streaming  context
    var ssc: StreamingContext = null

    if (args.length == 0) {
      model_file = "file:///Users/miaoyujia/git/mobile-userprofile-src/mprofile/file/base"
      basefeat2extentfeat_file = "file:///Users/miaoyujia/git/mobile-userprofile-src/mprofile/file/basefeat2extentfeat"
      config.setMaster("local[2]")
      if (sc == null) sc = new SparkContext(config)
    } else {
      model_file = args(0)
      basefeat2extentfeat_file = args(1)
      intervel = args(2).toInt
      if (sc == null) sc = new SparkContext(config)
    }
    log.info(MarkerFactory.getMarker("YuJia"), "Scoring, {}", Array("model_file", model_file, "basefeat2extentfeat_file", basefeat2extentfeat_file))

    val sMap = sc.textFile(basefeat2extentfeat_file).map {
      x =>
        var arr = x.split(",", -1)
        (arr(0), arr(1).toFloat)
    }.collectAsMap

    val muser = new MUser()
    for ((k, v) <- sMap) {
      println(k + "begin---------------")
      val t1 = System.currentTimeMillis()
      val rlist = hbaseServerce.isExist(muser, k)
      val t2 = System.currentTimeMillis()
      println("is in : " + rlist(0))
      val interval = t2 - t1
      println(k + "end---------------")
      println(k + "------------" + interval)
    }

    //    action_score.mapPartitions(existsAll).saveAsTextFile("/user/data-mprofile/tmp/readtest")
    //    action_score.map {
    //      case (k, v) =>
    //        println(k + "begin---------------")
    //        val t1 = System.currentTimeMillis()
    //        val muser = new MUser()
    //        val rlist = hbaseServerce.isExist(muser, k)
    //        (k, rlist(0))
    //        val t2 = System.currentTimeMillis()
    //        val interval = t2 - t1
    //        println(k + "end---------------")
    //        println(k + "------------" + interval)
    //    }.collect

  }

}
